﻿using System;
using System.Collections.Generic;
using System.Threading;
using CcrFlows;
using CcrFlows.Core;
using Microsoft.Ccr.Core;
using NUnit.Framework;

namespace Test.CcrFlows
{
    [TestFixture]
    public class testFlow
    {
        [Test]
        public void Create_1_stage_flow()
        {
            var seq = Flow<int>
                .Do<string>((i, ps) => ps.Post(i + "x"));

            string t;
            seq.Input.Post(1);
            Assert.IsTrue(seq.Output.WaitFor(500, out t));
            Assert.AreEqual("1x", t);
        }


        [Test]
        public void Receive_on_output()
        {
            var flow = Flow<int>.Do<int>(i => i + 1);
            flow.Input.Post(1);

            AutoResetEvent are = new AutoResetEvent(false);
            flow.Receive(i => are.Set());
            Assert.IsTrue(are.WaitOne(500));
        }


        [Test]
        public void Create_sequence()
        {
            var seq = Flow<int>
                .Do<int>((i, pi) => pi.Post(i + 1))
                .Do<string>((i, ps) => ps.Post(i + "x"))
                .Do<int>((s, ps) => ps.Post(s.Length));

            int n;
            seq.Input.Post(1);
            Assert.IsTrue(seq.Output.WaitFor(500, out n));
            Assert.AreEqual(2, n);
        }


        [Test]
        public void Create_fork()
        {
            var fork = Flow<int>.Fork<int, string>((i, pi, ps) => { pi.Post(i + 1);
                                                                      ps.Post(i + "x"); });
            fork.Input.Post(1);

            int n;
            string t;
            Assert.IsTrue(fork.Output0.WaitFor(500, out n));
            Assert.IsTrue(fork.Output1.WaitFor(500, out t));
            Assert.AreEqual(2, n);
            Assert.AreEqual("1x", t);
        }

       
        [Test]
        public void Create_fork_after_seq()
        {
            var flow = Flow<int>
                .Do<int>((i, pi)=>pi.Post(i+1))
                .Fork<int, string>((i, pi, ps) => { pi.Post(i + 1);
                                                      ps.Post(i+"x"); });
            flow.Input.Post(1);

            int n;
            string t;
            Assert.IsTrue(flow.Output0.WaitFor(500, out n));
            Assert.IsTrue(flow.Output1.WaitFor(500, out t));
            Assert.AreEqual(3, n);
            Assert.AreEqual("2x", t);
        }


        [Test]
        public void Creat_single_stage_parallel_sequence()
        {
            var fork = new ForkedFlow<int, int, string>((i, pi, ps) => { pi.Post(i + 1); ps.Post(i + "x"); })
                .Do<int, string>((i, pi) => pi.Post(i + 1), (s, ps) => ps.Post(s + "y"));

            fork.Input.Post(1);

            int n;
            string t;
            Assert.IsTrue(fork.Output0.WaitFor(500, out n));
            Assert.IsTrue(fork.Output1.WaitFor(500, out t));
            Assert.AreEqual(3, n);
            Assert.AreEqual("1xy", t);
        }


        [Test]
        public void Creat_multi_stage_parallel_sequence()
        {
            var fork = new ForkedFlow<int, int, string>((i, pi, ps) => { pi.Post(i + 1); ps.Post(i + "x"); })
                .Do<int, string>(
                    Flow<int>.Do<int>((i, pi) => pi.Post(i + 1)).Do<int>((i, pi) => pi.Post(i * 2)),
                    Flow<string>.Do<string>((s, ps) => ps.Post(s + "y")).Do<string>((s, ps) => ps.Post(s + "z"))
                    );

            fork.Input.Post(1);

            int n;
            string t;
            Assert.IsTrue(fork.Output0.WaitFor(500, out n));
            Assert.IsTrue(fork.Output1.WaitFor(500, out t));
            Assert.AreEqual(6, n);
            Assert.AreEqual("1xyz", t);

            fork = fork.Do<int, string>((i,pi)=>pi.Post(i+1), (s, ps)=>ps.Post(s+"a"));
            fork.Input.Post(1);
            Assert.IsTrue(fork.Output0.WaitFor(500, out n));
            Assert.IsTrue(fork.Output1.WaitFor(500, out t));
            Assert.AreEqual(7, n);
            Assert.AreEqual("1xyza", t);
        }


        [Test]
        public void Join_fork_immediately()
        {
            var flow = new ForkedFlow<int, int, string>((i, pi, ps) =>
                                                            {
                                                                pi.Post(i + 1);
                                                                ps.Post(i + "x");
                                                            })
                .Join<string>((i, s, ps) => ps.Post(i + "/" + s));
            flow.Input.Post(1);
            
            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.AreEqual("2/1x", t);
        }


        [Test]
        public void Join_after_parallel_flow()
        {
            var flow = new ForkedFlow<int, int, string>((i, pi, ps) =>
                            {
                                pi.Post(i + 1);
                                ps.Post(i + "x");
                            })
                .Do<int, string>((i, pi)=>pi.Post(i+1), (s, ps)=>ps.Post(s+"y"))
                .Join<string>((i, s, ps) => ps.Post(i + "/" + s));
            flow.Input.Post(1);

            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.AreEqual("3/1xy", t);
        }


        [Test]
        public void Follow_join_with_do()
        {
            var flow = new ForkedFlow<int>()
                .Join<int>((i0, i1, pi) => pi.Post(i0+i1))
                .Do<int>((i, pi) => pi.Post(i+1));
            
            flow.Input.Post(1);

            int n;
            Assert.IsTrue(flow.Output.WaitFor(500, out n));
            Assert.AreEqual(3, n);
        }



        [Test]
        public void Create_scatter_flow()
        {
            var scatter = Flow<int>.Scatter<string>((i, ps)=>ps.Post(i.ToString()));
            scatter.Input.Post(0);
            scatter.Input.Post(1);

            string t;
            Assert.IsTrue(scatter.Output.WaitFor(500, out t));
            Assert.IsTrue(scatter.Output.WaitFor(500, out t));
        }

        [Test]
        public void Append_scatter_to_flow()
        {
            var scatter = Flow<int>.Do<int>((i, pi)=>pi.Post(i+1)).Scatter<string>((i, ps) => ps.Post(i.ToString()));
            scatter.Input.Post(0);
            scatter.Input.Post(1);

            string t;
            Assert.IsTrue(scatter.Output.WaitFor(500, out t));
            Assert.IsTrue(scatter.Output.WaitFor(500, out t));
        }


        [Test]
        public void Concat_flows()
        {
            var flow0 = new Flow<int, int>((i, pi) => pi.Post(i + 1));
            var flow1 = new Flow<int, string>((i, ps) => ps.Post(i + "x"));
            var flow = flow0.Concat(flow1);

            flow.Input.Post(1);

            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.AreEqual("2x", t);
        }


        [Test]
        public void Use_simple_functions_as_handlers()
        {
            var flow = Flow<int>.Do<int>(i => i + 1).Do<string>(i=>i + "x");

            flow.Input.Post(1);

            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.AreEqual("2x", t);
        }

        [Test]
        public void Use_simple_func_as_join_handler()
        {
            var flow = new ForkedFlow<int>().Join<string>((i0, i1)=>i0+"/"+i1);

            flow.Input.Post(1);

            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.AreEqual("1/1", t);
        }

        [Test]
        public void Create_stateful_stage()
        {
            var count = new CounterStage {Counter = 0};
            var flow = new Flow<int, int>(count);

            flow.Input.Post(1);
            flow.Input.Post(2);
            flow.Input.Post(3);

            int n;
            Assert.IsTrue(flow.Output.WaitFor(500, out n));
            Assert.IsTrue(flow.Output.WaitFor(500, out n));
            Assert.IsTrue(flow.Output.WaitFor(500, out n));

            Assert.AreEqual(3, count.Counter);
        }


        [Test]
        public void Create_iterator_flow()
        {
            var flow = Flow<int>.DoWithIterator<string>(ProcessIntWithIterator);

            flow.Input.Post(3);

            string t;
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
            Assert.IsTrue(flow.Output.WaitFor(500, out t));
        }


        static IEnumerator<string> ProcessIntWithIterator(int input)
        {
            for (int i = 0; i < input; i++)
                yield return i.ToString();
        }
    }


    class CounterStage : ISeqHandler<int,int>
    {
        public int Counter;

        #region Implementation of ISeqHandler<int,int>

        public void Handle(int input, Port<int> output)
        {
            this.Counter++;
            output.Post(input);
        }

        #endregion
    }
}
